package com.gjy.kafka.sp.controller;

import com.gjy.kafka.sp.config.KafkaConstant;
import com.gjy.kafka.sp.entity.MessageEntity;
import com.gjy.kafka.sp.service.KafkaProducerService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * Kafka消息测试控制器
 * 提供API接口用于测试Kafka消息发送功能
 *
 * @author gjy
 * @version 1.0
 * @since 2025-09-07 10:48:21
 */
@RestController
@RequestMapping("/api/kafka")
public class KafkaMessageController {
    private static final Logger log = LoggerFactory.getLogger(KafkaMessageController.class);

    /**
     * Kafka消息生产者服务
     */
    private final KafkaProducerService kafkaProducerService;

    public KafkaMessageController(KafkaProducerService kafkaProducerService) {
        this.kafkaProducerService = kafkaProducerService;
    }

    /**
     * 发送普通消息
     *
     * @param content      消息内容
     * @param businessType 业务类型
     * @param businessId   业务ID
     * @param extra        额外信息
     * @return 响应信息
     */
    @PostMapping("/normal")
    public ResponseEntity<String> sendNormalMessage(
            @RequestParam String content,
            @RequestParam(required = false) String businessType,
            @RequestParam(required = false) String businessId,
            @RequestParam(required = false) String extra) {

        log.info("接收到发送普通消息请求");
        MessageEntity message = kafkaProducerService.createMessageEntity(content, businessType, businessId, extra);
        kafkaProducerService.sendMessage(KafkaConstant.NORMAL_TOPIC, message);
        return ResponseEntity.ok("普通消息发送成功，消息ID：" + message.getMessageId());
    }

    /**
     * 发送分区消息
     *
     * @param content      消息内容
     * @param businessType 业务类型
     * @param businessId   业务ID（用于分区）
     * @param extra        额外信息
     * @return 响应信息
     */
    @PostMapping("/partition")
    public ResponseEntity<String> sendPartitionMessage(
            @RequestParam String content,
            @RequestParam(required = false) String businessType,
            @RequestParam String businessId,
            @RequestParam(required = false) String extra) {

        log.info("接收到发送分区消息请求，业务ID：{}", businessId);
        MessageEntity message = kafkaProducerService.createMessageEntity(content, businessType, businessId, extra);
        kafkaProducerService.sendPartitionMessage(message);
        return ResponseEntity.ok("分区消息发送成功，消息ID：" + message.getMessageId());
    }

    /**
     * 发送事务消息
     *
     * @param content      消息内容
     * @param businessType 业务类型
     * @param businessId   业务ID
     * @param extra        额外信息
     * @return 响应信息
     */
    @PostMapping("/transactional")
    public ResponseEntity<String> sendTransactionalMessage(
            @RequestParam String content,
            @RequestParam(required = false) String businessType,
            @RequestParam(required = false) String businessId,
            @RequestParam(required = false) String extra) {

        log.info("接收到发送事务消息请求");
        MessageEntity message = kafkaProducerService.createMessageEntity(content, businessType, businessId, extra);
        kafkaProducerService.sendTransactionalMessage(message);
        return ResponseEntity.ok("事务消息发送成功，消息ID：" + message.getMessageId());
    }

}
